package com.chris.core.socket.bio.util;

import com.chris.core.socket.bio.dto.SocketMapDto;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.net.*;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.*;

/**
 * 多线程，同步阻塞式 -> sokect 服务端
 * 抽象类
 * 苏斌 2022-3-19
 */
public abstract class BioSokcetServerUtil {
    //#region 参数及变量
    /**
     * 服务端
     */
    ServerSocket serverSocket = null;
    private Logger log = LoggerFactory.getLogger(this.getClass());
    /**
     * 线程池
     * 1. newCachedThreadPool 创建一个可缓存的线程池，调用execute 将重用以前构造的线程（如果线程可用）。如果没有可用的线程，则创建一个新线程并添加到池中。终止并从缓存中移除那些已有 60 秒钟未被使用的线程。
     * 2. newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
     * 3. newScheduledThreadPool 创建一个定长线程池，支持定时及周期性任务执行。
     * 4. newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
     */
    private ExecutorService executorService =  Executors.newCachedThreadPool();//Executors.newFixedThreadPool(30000);//
    /**
     * ScheduledThreadPoolExecutor是ExecutorService的另一个实现类,ScheduledThreadPoolExecutor 类的功能也主要体现在ScheduledExecutorService 接口上.
     */
    //private ScheduledExecutorService worker = Executors.newSingleThreadScheduledExecutor();//Executors.newScheduledThreadPool(1);
    /**
     * 客服端hash集合
     */
    private ConcurrentHashMap<String, SocketMapDto> socketMap = new ConcurrentHashMap<>();
    /**
     * 客服端在线数
     */
    private int onlineCount = 0;

    //private String host;
    private int port;

    /**
     * 最大socket线程数
     */
    private int socketCount;

    /**
     * 消息断言
     */
    private String breakMark = "\r\n";// \n    \r\n    ~!@#bbq8$%^&*\n;

    /**
     * 回车  chr(13) \r
     * readline 是否包含,结合 breakMark 一起判断
     */
    private String _CR="\r";

    /**
     * 回车  chr(10) \n
     * readline 是否包含,结合 breakMark 一起判断
     */
    private String _LF="\n";

    /**
     * 读取超时
     */
    private int readTimeout=181000;

    /**
     * 上传文件夹
     */
    private String uploadFilePath="./uploads/";

    public String getUploadFilePath() {
        return uploadFilePath;
    }

    public void setUploadFilePath(String uploadFilePath) {
        this.uploadFilePath = uploadFilePath;
    }

    //#endregion

    //#region 初始化系统
    /**
     * 够造函数
     *
     * @param port     监听端口
     * @param breakMark 消息断言
     */
    public BioSokcetServerUtil(int port, String breakMark) {
        this.port=port;
        this.socketCount=25000;
        this.breakMark=breakMark;
    }


    /**
     * 够造函数
     *
     * @param port     监听端口
     * @param socketCount   最大线程数
     * @param readTimeout   读超时
     * @param breakMark 消息断言 默认 \r\n , 如果为空，即是不间断读写，有数据就抛出
     */
    public BioSokcetServerUtil(int port,int socketCount,int readTimeout, String breakMark) {
        this.port=port;
        this.readTimeout=readTimeout;
        this.socketCount=socketCount;
        this.breakMark=breakMark;
    }

    /**
     * 够造函数
     *
     * @param port 监听端口
     * @throws IOException
     */
    public BioSokcetServerUtil(int port) {
        this.port=port;
        this.socketCount=25000;
    }

    /**
     * 够造函数
     *
     * @param port 监听端口
     * @throws IOException
     */
    public BioSokcetServerUtil(int port,int socketCount) {
        this.port=port;
        this.socketCount=socketCount;
    }


    /**
     * 初始化系统
     *
     */
    private void IntSystem() {
        if (breakMark.endsWith("\r\n"))
        {
            _CR="\r";
            _LF="\n";
        }
        else if (breakMark.endsWith("\n"))
        {
            _CR="";
            _LF="\n";
        }
        else
        {
            //目前采用readline方式,此处会出错
            //2022-7-13修正,可接收不间段数据流
            _CR="";
            _LF="";
        }

    }

   public void Open()
   {
       Thread t = new Thread(new Runnable() {
           @Override
           public void run() {
               try {
                   Thread.sleep(3000);
                   IntSystem();
                   bind();

               } catch (IOException e) {
                   e.printStackTrace();
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
       });
       t.start();
   }

    /**
     * 打开链接
     *
     * @throws IOException
     */
    private void bind() throws IOException, InterruptedException {


        serverSocket = new ServerSocket();//port
        // 是否复用未完全关闭的地址端口
        //serverSocket.setReuseAddress(true);
        // 等效Socket#setReceiveBufferSize
        //不管发送端和接收端缓冲区的大小如何设置，最后生效的是两者之间值最小的那个大小
      //  int i=serverSocket.getReceiveBufferSize();
        serverSocket.setReceiveBufferSize(serverSocket.getReceiveBufferSize()*2);

        // 设定阻塞时间--> 针对 serverSocket.accept()--有时候有可能 accept 很慢会卡死，必须设置超时调试，否则整个程序卡死
       // serverSocket.setSoTimeout(15000);

        // 绑定到本地端口上 要前面设置生效，bind命令后面执行
        //backlog -》.accept() 前 缓存队列
        serverSocket.bind(new InetSocketAddress( port), 128);
        log.info("启动BIO Socket 服务监听 端口:"+port);
        //流量缓冲区大小
        // socket.setReceiveBufferSize();
        // socket.setSendBufferSize();
        while (true) {
            //大于最大连接数不接受新的连接
            if (socketCount <= getOnlineCount())// || true
            {
                Thread.sleep(1000);
                continue;
            }
            //限制:每秒100个链接
            Thread.sleep(10);
            //流量控制
            //等待客服端连接
            //accept阻塞式等待

            Socket socket = null;
            //生成socket必须用同步方式，如果前方多线程并发，serverSocket.accept();可能会卡死
            try {
                socket = serverSocket.accept();
              //  log.debug("接单Socket链接 -> 总客服端数：" + getOnlineCount());
            }
            catch (Exception ex)
            {
                //accept 超时，重新来
                log.debug("serverSocket.accept超时"+ex.toString());
                socket=null;
               // Thread.sleep(3000);
                continue;
            }
            //不管发送端和接收端缓冲区的大小如何设置，最后生效的是两者之间值最小的那个大小
            socket.setReceiveBufferSize(socket.getReceiveBufferSize()*2);
            socket.setSendBufferSize(socket.getSendBufferSize()*2);
            socket.setKeepAlive(true);
            //读取等待超时，超过30秒没有数据流自动断开,要设置心跳
            //主要针对  socket--》read() (服务端的，非客户端） ,注意和 serverSocket.setSoTimeout 区别
            socket.setSoTimeout(readTimeout);
            String uuid = UUID.randomUUID().toString();
            SocketMapDto dto = new SocketMapDto(uuid, socket);
            addOnlineCount();
            socketMap.put(uuid, dto);
            //DoReviceMsg(dto);
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    DoReviceMsg(dto);
                }
            });
            log.debug("建立客服端Socket链接 id:" + dto.getCid() + "-> 总客服端数：" + getOnlineCount());
            //时间间隔开启
            // Thread.sleep(10);

        }
    }

    /**
     * 关闭链接
     *
     * @param sid 客服端uuid
     */
    public void Stop(String sid) {
        SocketMapDto dto = socketMap.get(sid);
        if (dto != null && dto.getSocket() != null) {
            try {
                Socket socket = dto.getSocket();
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            // socketMap.remove(dto);
            // subOnlineCount();
            log.debug("服务端关闭socket连接:" + dto.getCid());
        }
    }

    //#endregion

    //#region 接收内容


    /**
     * 循环接收消息(文本）
     * @param dto
     */
    private void DoReviceMsg(SocketMapDto dto) {
       // log.debug("开始循环获取Socke id:" + dto.getCid() + "-> 总客服端数：" + getOnlineCount());
        InputStream inputStream = null;
        BufferedReader bufferedReader = null;
        // StringBuffer sbf = new StringBuffer();
        Socket socket =dto.getSocket();
        try {
            //-----------方法1-------------
            inputStream = socket.getInputStream();
            //循环读取客服端发送数据
            if (_CR=="" && _LF=="") {
                //没有断言情况，不间断读写 ，有数据流就抛出 2022-7-13
                byte[] bytes = new byte[1024];
                int len;
                StringBuilder sb = new StringBuilder();
                //只有当客户端关闭它的输出流的时候，服务端才能取得结尾的-1
                while ((len = inputStream.read(bytes)) != -1) {
                    // 注意指定编码格式，发送方和接收方一定要统一，建议使用UTF-8
                    sb.append(new String(bytes, 0, len, "UTF-8"));
                    OnRevice(dto.getCid(), sb.toString());
                }
            }
            else {
                //有断言的情况按行读取
                //由Socket对象得到输入流，并构造相应的BufferedReader对象
                bufferedReader = new BufferedReader(new InputStreamReader(inputStream, "UTF-8"));
                //临时读入文本
                String readline;
                //文本缓冲
                StringBuilder stringBuilder = new StringBuilder();
                do {
                    // log.debug("等待读取数据，阻塞 线程:" + Thread.currentThread().getId() + " " + Thread.currentThread().getName());
                    readline = bufferedReader.readLine(); //从系统标准输入读入一字符串
                    if (readline != null) {
                        readline += _CR + _LF;  //readLine()去掉了原始\n 或 \r\n ,重新加上恢复原始字符串内容
                        if (readline.equals("!@#$%cmd:file:begin^&*" + breakMark)) {
                            log.debug("截获上传文件命令，转跳上传文件");
                            //第二行接收文件名
                            String fileName = delSuffixMsg(bufferedReader.readLine() + _CR + _LF);
                            String str = "";
                            str = delSuffixMsg(bufferedReader.readLine() + _CR + _LF);
                            //第三方接收文件大小
                            Long fileSize = Long.parseLong(str);
                            DoReviceFile(fileName, fileSize, socket);

                            continue;//继续下一个循环获取消息
                        }

                        if (readline.endsWith(breakMark))//注意，如果参数是空字符串，或者等于此 String 对象（用 equals(Object) 方法确定），则结果为 true
                        {
                            //如果存在断言-->单独一行内有断言，为之结束
                            readline = delSuffixMsg(readline);
                            stringBuilder.append(readline);
                            if (stringBuilder.length() > 0) {
                                OnRevice(dto.getCid(), stringBuilder.toString());
                                stringBuilder.setLength(0);
                            }
                        } else {
                            //没有断言， 当作内容添加进stringBuilder
                            stringBuilder.append(readline);
                        }
                    /*
                    if (breakMark.equals("\n")) {
                        //每行是一个节点，内容内不能有\n
                        OnRevice(dto.getCid(), readline);
                        stringBuilder.setLength(0);
                    } else {


                        if (readline.equals(breakMark)) {
                            if (stringBuilder.length()>0) {
                                //如果存在断言-->单独一行内有断言，为之结束
                                OnRevice(dto.getCid(), stringBuilder.toString());
                                stringBuilder.setLength(0);
                            }
                        } else {
                            //没有断言，\n 当作内容添加进stringBuilder
                            stringBuilder.append(readline+"\n");
                        }
                    }*/
                    }

                }
                while (readline != null);//&& !readline.equals(breakMark)//有坑！！！！！===》 readLine 必须是有 \n  才会读取完毕
            }
            //----------方法2-------------
            /*
            byte[] bytes = new byte[1024];
            int len;
            StringBuilder sb = new StringBuilder();
            //只有当客户端关闭它的输出流的时候，服务端才能取得结尾的-1
            while ((len = inputStream.read(bytes)) != -1) {
                // 注意指定编码格式，发送方和接收方一定要统一，建议使用UTF-8
                sb.append(new String(bytes, 0, len, "UTF-8"));
            }
            System.out.println("get message from client: " + sb);

             */
            //----------方法3------------
            /*
            //注意此方法有限制：用dataOutputStream写入，要用 DataInputStream 读取，因为
            //dataOutputStream 写入内容，带有文件类型识别前缀，需要DataInputStream识别
            inputStream = dto.getSocket().getInputStream();
            DataInputStream dataInputStream = new DataInputStream(inputStream);
            //由Socket对象得到输入流，并构造相应的BufferedReader对象
            //临时读入文本
            String readline;

            //循环读取客服端发送数据
            do {
                // log.debug("等待读取数据，阻塞 线程:" + Thread.currentThread().getId() + " " + Thread.currentThread().getName());
                readline = dataInputStream.readUTF(); //从系统标准输入读入一字符串
                if (readline != null) {
                    OnRevice(dto.getCid(), readline);
                }
            }
            while (readline != null && !readline.equals(breakMark));//有坑！！！！！===》 readLine 必须是有 \n  才会读取完毕
            */
            //------------------------

        } catch (Exception ex) {
            log.debug("服务端读->取内容断开" + ex.toString());
            // ex.printStackTrace();
        } finally {
            if (inputStream != null) {
                try {
                    inputStream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            if (bufferedReader != null) {
                try {
                    bufferedReader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            ;
            if (socket != null)
                try {
                    socket.close();
                    dto.setSocket(null);

                } catch (IOException e) {
                    e.printStackTrace();
                }
            socketMap.remove(dto.getCid());
            subOnlineCount();
            log.debug("服务端Socket链接关闭 id:" + dto.getCid() + "-> 总客服端数：" + getOnlineCount());
        }

    }

    /**
     * 接收文件
     * 仅获取上传文件动作时候可用
     * @param fileName
     * @param socket
     */
    private void DoReviceFile(String fileName,Long fileSize,Socket socket)
    {
        FileOutputStream fos=null;
        DataInputStream  dis=null;
        try {
              dis = new DataInputStream(socket.getInputStream());
            // 文件名和长度
           // String fileName = dis.readUTF();
            //readLong有问题，亏阻断，原因不明
           // long fileLength = dis.readLong();
            log.debug("文件的长度为:" + fileSize);
            File directory = new File(uploadFilePath);
            if(!directory.exists()) {
                directory.mkdir();
            }
            File file = new File(directory.getAbsolutePath() + File.separatorChar + fileName);
           if (file!=null && file.isFile() && file.exists() )
               file.delete();
            fos = new FileOutputStream(file);
            log.debug("file。。。。。。。。。。。。。。"+file);
            log.debug("fileName。。。。。。。。。。。。。。"+fileName);
            log.debug("======== 开始接收文件 ========");
            byte[] bytes = new byte[1024];
            int length = 0;
            long passedlen=0;
            Long per=0L;
            //统计开始时间
            long beginTime=System.currentTimeMillis();

            //注意坑，不能做死循环 =-1 ，因为从 外部while调入进来的，内部死循环出不去
            //外部就永远收不到消息了，只能定长读取内容
            while((length = dis.read(bytes, 0, bytes.length)) != -1) {

                fos.write(bytes, 0, length);
                fos.flush();
                passedlen+=length;
                if (per!=passedlen*100/fileSize) {
                    per=passedlen*100/fileSize;
                    SendClientMsg("!@#$%file:send^&*文件接收了:" + per + "%", socket);
                }
                //如果已经接到完文件，直接跳出，不while
                if (passedlen>=fileSize)//收到正确的流文件跳出
                    break;

                //流量控制
                flowControl(passedlen,beginTime,200);
            }
            SendClientMsg("!@#$%file:end^&*文件接收完毕", socket);
            log.debug("======== 文件接收成功 耗时:"+(System.currentTimeMillis()-beginTime)/1000+"秒========");

        } catch (Exception e) {
            log.error("上传文件["+fileName+"]出错:"+e.toString());
            File directory = new File(uploadFilePath);
            File file = new File(directory.getAbsolutePath() + File.separatorChar + fileName);
            if (file!=null && file.isFile() && file.exists())
            {
                //上传文件失败，删除文件
                file.delete();
            }
            // e.printStackTrace();
        } finally {
            try {
                //可能会关闭socket注意后面调试测试
                if(fos != null)
                    fos.close();
               // if(dis != null)
               //     dis.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 循环接收消息(字节）
     *
     * @param dto
     */
    private void DoReviceBytes(SocketMapDto dto) {


    }

    /**
     * 循环接收消息(16H）->暂时是空方法
     *
     * @param dto
     */
    private void DoReviceHex(SocketMapDto dto) {

    }

    //#endregion

    //#region 外抛抽象事件
    /**
     * 抽象获取消息
     * 接收文本消息有用
     *
     * @param sid 客服端唯一ID
     * @param mag 获取消息
     */
    public abstract void OnRevice(String sid, String mag);

    //#endregion

    //#region 发送消息及其命令

    /**
     * 抽象获取消息
     * 接收字节流数据有用
     */
    //public abstract void OnReviceBytes(String sid,byte[] bytes);

    //#endre
    /**
     * 服务端向客服端发送信息
     *
     * @param msg 发给客服端消息
     * @param sid 客服端id
     */
    public void SendClientMsg(String msg, String sid) throws IOException {
        SocketMapDto dto = socketMap.get(sid);
        Socket socket = dto.getSocket();
        if (dto == null || socket == null || !socket.isConnected()) {
            log.error("客服端不存在,或未连接");
            throw new IOException("客服端不存在,或未连接");
        }
        SendClientMsg(msg, socket);
    }

    /**
     * 服务端向客服端发送消息
     *
     * @param msg    发给客服端消息
     * @param socket 客服端socket
     * @throws IOException
     */
    public void SendClientMsg(String msg, Socket socket) throws IOException {
        //多线程，本实例内，全局锁定【socket】，排队轮流发送 消息 或文件
        synchronized(socket) {
            if (socket == null || !socket.isConnected()) {
                log.error("客服端不存在,或未连接");
                throw new IOException("客服端不存在,或未连接");
            }

            if (socket != null && socket.isConnected()) {
                OutputStream outputStream = null;
                // DataOutputStream dataOutputStream = null;

                try {
                    outputStream = socket.getOutputStream();
                    //------------方法1--------------
                    msg += breakMark;
                    outputStream.write(msg.getBytes(StandardCharsets.UTF_8));
                    //-----------方法2--------------------
                    //由Socket对象得到输出流，并构造PrintWriter对象
                /*
                PrintWriter os=new PrintWriter(outputStream);
                if (breakMark.equals("\n"))
                    msg += "\n";
                else
                    msg += "\n" + breakMark;
                 os.write(msg);
                os.flush();

                 */
                    //----------方法3-----------------
                /*
                //注明：有限制 用dataOutputStream写入，要用 DataInputStream 读取，因为
                //dataOutputStream 写入内容，带有文件类型识别前缀，需要DataInputStream识别
                dataOutputStream = new DataOutputStream(outputStream);
                if (breakMark.equals("\n"))
                    msg += "\n";
                else
                    msg += "\n" + breakMark;
                 dataOutputStream.writeUTF(msg);

                dataOutputStream.flush();
                */
                } catch (IOException e) {
                    e.printStackTrace();
                } finally {
              /*
                if (outputStream!=null) {
                    try {
                        outputStream.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }

                if (dataOutputStream!=null) {
                    try {
                        dataOutputStream.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }*/
                }

            }
        }
    }

    /**
     * 群发所有消息
     *
     * @param msg
     * @throws IOException
     */
    public void SendAllClientMsg(String msg) throws IOException {
        for (SocketMapDto dto : socketMap.values()
        ) {
            SendClientMsg(msg, dto.getSocket());
        }
    }

    /**
     *发送文件到客服端
     *
     * @param sid 客服端id
     */
    public void SendClientMsg(String fileName, File file, String sid) throws IOException {
        SocketMapDto dto = socketMap.get(sid);
        Socket socket = dto.getSocket();
        if (dto == null || socket == null || !socket.isConnected()) {
            log.error("客服端不存在,或未连接");
            throw new IOException("客服端不存在,或未连接");
        }
        SendClientFile(fileName,file, socket);
    }

    /**
     * 发送文件到客服端
     *
     * @param fileName    文件名称
     * @param file    文件
     */
    public void SendClientFile(String fileName, File file,Socket socket) {
        if (file==null || !file.isFile() || !file.exists())
            return;
        OutputStream outputStream = null;
        InputStream file_inputStream = null;
        //多线程，本实例内，全局锁定【socket】，排队轮流发送 消息 或文件
        synchronized(socket) {
            if (socket != null && socket.isConnected()) {
                try {
                    outputStream = socket.getOutputStream();
                    //------------方法1--------------
                    //启动命令，消息转跳动作
                    SendClientMsg("!@#$%cmd:file:begin^&*",socket);
                    //文件名称
                    SendClientMsg(fileName,socket);
                    //文件大小
                    SendClientMsg(file.length() + "",socket);
                    // 开始传输文件
                    log.debug("======== 开始传输文件 ========");
                    byte[] bytes = new byte[1024];
                    int length = 0;
                    long passedlen=0;
                    //统计开始时间
                    long beginTime=System.currentTimeMillis();
                    file_inputStream = new FileInputStream(file);
                    while ((length = file_inputStream.read(bytes, 0, bytes.length)) != -1) {
                        outputStream.write(bytes, 0, length);
                        outputStream.flush();

                        //流量控制
                        passedlen+=length;
                        flowControl(passedlen,beginTime,200);
                    }
                    log.debug("======== 文件传输成功 耗时:"+(System.currentTimeMillis()-beginTime)/1000+"秒========");
                    //SendMsg("!@#$%cmd:file:end^&*");
                    // outputStream.write(msg.getBytes(StandardCharsets.UTF_8));
                } catch (IOException e) {
                    log.error("上传文件错误:" + e.toString());
                    //e.printStackTrace();
                }  finally {
                    if (file_inputStream != null) {
                        try {
                            file_inputStream.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }

                /*
                不能关闭，关闭整个socket就关了
                if (outputStream != null) {
                    try {
                        outputStream.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }

                 */
                }

            }
        }
    }
    //#endregion

    //#region 线程同步 及其他

    public synchronized int getOnlineCount() {
        return onlineCount;
    }

    public synchronized void addOnlineCount() {
        onlineCount++;
    }

    public synchronized void subOnlineCount() {
        onlineCount--;
    }

    /**
     * 流量控制  (目前策略 接收是 发送 2 被 限流 ，双方需要约定好 ）
     * 这是接收端端限制，因为是并发异步/同步，所以不能同步阻止发送端写速（NIO/BIO）
     * 只能总限制 并发总数 * 单并发 = 总速度 , 例如 ： 限制 200KB * 100 = 20 M / S  ,1000并发 200M， 1W并发 2000M
     * @param passedlen 总已传输字节 (b！,非KB）
     * @param beginTime 开始时间
     * @param limitspped 限制速度  kb / s ( 200 = 200KB/s , 2 Kb/ms)
     */
    private void flowControl(long passedlen ,long beginTime,long limitspped)
    {
        //流量控制 100KB/S = 100  * 1024  b/s  = 1024000 b/S =  102 b /ms
        //limitspped KB /s = limitspped * 1024 * 1000 / 1000 = limitspped * 1024  / ms
        long passTime = (System.currentTimeMillis() - beginTime);
        if(passTime==0) passTime=1;
        long pp = passedlen/limitspped - passTime ;//如果passTime=0，按实际所需passedlen/limitspped等待
        if (pp > 0) {
            try {
                Thread.sleep(pp);

            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 去除字符串后缀
     * @param msg
     * @return
     */
    private String delSuffixMsg(String msg) {
        if (msg.endsWith(breakMark)) {
            msg= msg.substring(0,msg.length()-breakMark.length());
        }
        return msg;
    }
    //#endregion


}
